package com.fengplus.demo.consumer;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.UUID;

@Component
@Slf4j
public class RabbitmqConsumer {

    // #配置手动签收
    // spring.rabbitmq.listener.simple.acknowledge-mode=manual
    // spring.rabbitmq.listener.direct.acknowledge-mode=manual
    // #消息预读数量 1表示每次从队列中读取一条消息
    // spring.rabbitmq.listener.simple.prefetch=1

    // Hello模式：发布-消费，自动确认
    @RabbitListener(queues = "queue.a")
    public void ConsumerA(String msg, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
                          Channel channel) throws IOException {
        try {
            //模拟出现异常
//            int a = 1 / 0;
            log.info("收到的消息-> msg:{}", msg);
            log.info("收到的消息标识-> deliveryTag:{}", deliveryTag);
            log.info("收到的消息-> channel:{}", channel.toString());
            //手动签收  channel.basicAck(消息唯一标识,是否批量签收);
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            e.printStackTrace();
            //channel.basicNack(deliveryTag:消息的唯一标识,multiple:是否批量处理,requeue:是否重新放入队列);
            //消息出现异常时，若requeue=false，则该消息会被放入死信队列，若没有配置死信队列则该消息会丢失。
            channel.basicNack(deliveryTag, false, false);
        }
    }

    // Work模式：手动确认、每次接收1条、消费异常处理
    @RabbitListener(
        queuesToDeclare = @Queue("queue.b")
    )
    public void ConsumerB(String msg, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
                          Channel channel) throws IOException {

        try {
            //模拟出现异常
//            int a = 1 / 0;
            log.info("收到的消息-> msg:{}", msg);
            log.info("收到的消息标识-> deliveryTag:{}", deliveryTag);
            log.info("收到的消息-> channel:{}", channel.toString());
            //手动签收  channel.basicAck(消息唯一标识,是否批量签收);
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            e.printStackTrace();
            //channel.basicNack(deliveryTag:消息的唯一标识,multiple:是否批量处理,requeue:是否重新放入队列);
            //消息出现异常时，若requeue=false，则该消息会被放入死信队列，若没有配置死信队列则该消息会丢失。
            channel.basicNack(deliveryTag, false, false);
        }
    }

    // Publish/Subscribe模式
    //创建队列和交换机的绑定关系
    //创建个匿名队列
    //定义交换机，指定交换机名称和fanout类型
    @RabbitListener(
        bindings = @QueueBinding(
            value = @Queue(),
            exchange = @Exchange(name = "exchange.c", type = "fanout")
        )
    )
    public void ConsumerC1(String msg, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
                           Channel channel) throws IOException {
        try {
            log.info("01 消费者收到的消息 -> msg:{}", msg);
            //手动签收  channel.basicAck(消息唯一标识,是否批量签收);
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            e.printStackTrace();
            //channel.basicNack(deliveryTag:消息的唯一标识,multiple:是否批量处理,requeue:是否重新放入队列);
            //消息出现异常时，若requeue=false，则该消息会被放入死信队列，若没有配置死信队列则该消息会丢失。
            channel.basicNack(deliveryTag, false, false);
        }
    }

    @RabbitListener(
        bindings = @QueueBinding(
            value = @Queue(),
            exchange = @Exchange(name = "exchange.c", type = "fanout")
        )
    )
    public void ConsumerC2(String msg, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
                           Channel channel) throws IOException {
        try {
            log.info("02 消费者收到的消息 -> msg:{}", msg);
            //手动签收  channel.basicAck(消息唯一标识,是否批量签收);
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            e.printStackTrace();
            //channel.basicNack(deliveryTag:消息的唯一标识,multiple:是否批量处理,requeue:是否重新放入队列);
            //消息出现异常时，若requeue=false，则该消息会被放入死信队列，若没有配置死信队列则该消息会丢失。
            channel.basicNack(deliveryTag, false, false);
        }
    }

    // Routing模式
    @RabbitListener(
        bindings = @QueueBinding(
            value = @Queue(),
            exchange = @Exchange(name = "exchange.d", type = "direct"),
            key = {"route.d.error", "route.d.info", "route.d.warning"}
        )
    )
    public void ConsumerD1(String msg,
                           @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
                           Channel channel) throws IOException {
        try {
            log.info("01 消费者收到的消息 -> msg:{}", msg);
            //手动签收  channel.basicAck(消息唯一标识,是否批量签收);
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            e.printStackTrace();
            //channel.basicNack(deliveryTag:消息的唯一标识,multiple:是否批量处理,requeue:是否重新放入队列);
            //消息出现异常时，若requeue=false，则该消息会被放入死信队列，若没有配置死信队列则该消息会丢失。
            channel.basicNack(deliveryTag, false, false);
        }
    }

    @RabbitListener(
        bindings = @QueueBinding(
            value = @Queue(),
            exchange = @Exchange(name = "exchange.d", type = "direct"),
            key = {"route.d.error"}
        )
    )
    public void ConsumerD2(String msg,
                           @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, //@Header 获取消息体header中的信息
                           Channel channel) throws IOException {
        try {
            log.info("02 消费者收到的消息 -> msg:{}", msg);
            //手动签收  channel.basicAck(消息唯一标识,是否批量签收);
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            e.printStackTrace();
            //channel.basicNack(deliveryTag:消息的唯一标识,multiple:是否批量处理,requeue:是否重新放入队列);
            //消息出现异常时，若requeue=false，则该消息会被放入死信队列，若没有配置死信队列则该消息会丢失。
            channel.basicNack(deliveryTag, false, false);
        }
    }

    // Topics模式
    @RabbitListener(
        bindings = @QueueBinding(
            value = @Queue(),
            exchange = @Exchange(name = "exchange.e", type = "topic"),
            key = {"route.e.error"}
        )
    )
    public void ConsumerE1(String msg,
                           @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, //@Header 获取消息体header中的信息
                           Channel channel) throws IOException {
        try {
            log.info("01 消费者收到的消息 -> msg:{}", msg);
            //手动签收  channel.basicAck(消息唯一标识,是否批量签收);
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            e.printStackTrace();
            //channel.basicNack(deliveryTag:消息的唯一标识,multiple:是否批量处理,requeue:是否重新放入队列);
            //消息出现异常时，若requeue=false，则该消息会被放入死信队列，若没有配置死信队列则该消息会丢失。
            channel.basicNack(deliveryTag, false, false);
        }
    }

    @RabbitListener(
        bindings = @QueueBinding(
            value = @Queue(),
            exchange = @Exchange(name = "exchange.e", type = "topic"),
            key = {"route.e.*"}
        )
    )
    public void ConsumerE2(String msg,
                           @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, //@Header 获取消息体header中的信息
                           Channel channel) throws IOException {
        try {
            log.info("02 消费者收到的消息 -> msg:{}", msg);
            //手动签收  channel.basicAck(消息唯一标识,是否批量签收);
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            e.printStackTrace();
            //channel.basicNack(deliveryTag:消息的唯一标识,multiple:是否批量处理,requeue:是否重新放入队列);
            //消息出现异常时，若requeue=false，则该消息会被放入死信队列，若没有配置死信队列则该消息会丢失。
            channel.basicNack(deliveryTag, false, false);
        }
    }

    @RabbitListener(
        bindings = @QueueBinding(
            value = @Queue(),
            exchange = @Exchange(name = "exchange.e", type = "topic"),
            key = {"route.#"}
        )
    )
    public void ConsumerE3(String msg,
                           @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, //@Header 获取消息体header中的信息
                           Channel channel) throws IOException {
        try {
            log.info("03 消费者收到的消息 -> msg:{}", msg);
            //手动签收  channel.basicAck(消息唯一标识,是否批量签收);
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            e.printStackTrace();
            //channel.basicNack(deliveryTag:消息的唯一标识,multiple:是否批量处理,requeue:是否重新放入队列);
            //消息出现异常时，若requeue=false，则该消息会被放入死信队列，若没有配置死信队列则该消息会丢失。
            channel.basicNack(deliveryTag, false, false);
        }
    }

    @RabbitListener(queues = "route.f.rpc")
    public String ConsumerFRpc(String msg,
                               @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
                               @Header(AmqpHeaders.CORRELATION_ID) String correlationId,
                               Channel channel) throws IOException {
        try {
            log.info("RPC 收到msg:{}", msg);
            //手动签收  channel.basicAck(消息唯一标识,是否批量签收);
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            e.printStackTrace();
            //channel.basicNack(deliveryTag:消息的唯一标识,multiple:是否批量处理,requeue:是否重新放入队列);
            //消息出现异常时，若requeue=false，则该消息会被放入死信队列，若没有配置死信队列则该消息会丢失。
            channel.basicNack(deliveryTag, false, false);
        }
        String returnMsg = UUID.randomUUID().toString();
        log.info("RPC 收到correlationId:{}", correlationId);
        log.info("RPC 返回msg:{}", returnMsg);
        return returnMsg;
    }

    @RabbitListener(
        bindings = @QueueBinding(
            value = @Queue(),
            exchange = @Exchange(name = "exchange.g"),
            key = {"route.g"}
        )
    )
    public void ConsumerG(String msg,
                           @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, //@Header 获取消息体header中的信息
                           Channel channel) throws IOException {
        try {
            log.info("收到的消息-> msg:{}", msg);
            log.info("收到的消息标识-> deliveryTag:{}", deliveryTag);
            log.info("收到的消息-> channel:{}", channel.toString());
            //手动签收  channel.basicAck(消息唯一标识,是否批量签收);
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            e.printStackTrace();
            //channel.basicNack(deliveryTag:消息的唯一标识,multiple:是否批量处理,requeue:是否重新放入队列);
            //消息出现异常时，若requeue=false，则该消息会被放入死信队列，若没有配置死信队列则该消息会丢失。
            channel.basicNack(deliveryTag, false, false);
        }
    }

}
